package com.tgy.twodatabaseplus.kafka;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.time.temporal.ChronoField;
import java.util.*;

/***
 * @ClassName: TgyConsumer
 * @Description:
 * @Auther: 送你一块二向箔
 * @Date: 17:33 2022/4/2
 * @version : V1.0
 */
@Slf4j
@Component
public class TgyConsumer {
    public static void main(String[] args) {
        resver("sync");
    }
    private final static String TOPIC_NAME = "yaml";
    private final static String CONSUMER_GROUP_NAME = "bbb";

    public static void resver(String flag){
        //同步还是异步提交
        //String flag = flag;
        Properties props = new Properties();
        //是否自动提交
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false");
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"47.101.193.71:9092");
        // 消费分组名
        props.put(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP_NAME);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
        //创建一个消费者的客户端
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String,String>(props);
        // 消费者订阅主题列表
        //consumer.subscribe(Arrays.asList(TOPIC_NAME));

        //指定分区消费
        //consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0 )));
        //消息回溯消费
        //consumer.seekToBeginning(Arrays.asList(new TopicPartition(TOPIC_NAME,0 )));
        //指定offset消费
        //consumer.seek(new TopicPartition(TOPIC_NAME, 0 ), 10 );


        List<PartitionInfo> topicPartitions =consumer.partitionsFor(TOPIC_NAME);
        //从 1 小时前开始消费
        long milliSecond = LocalDateTime.now().toInstant(ZoneOffset.of("+8")).toEpochMilli() - 1800000;
        long time = System.currentTimeMillis() - 3600000;
        Map<TopicPartition, Long> map = new HashMap<>();
        for (PartitionInfo par : topicPartitions) {
            map.put(new TopicPartition(TOPIC_NAME, par.partition()),milliSecond);
        }
        Map<TopicPartition, OffsetAndTimestamp> parMap =consumer.offsetsForTimes(map);
        for (Map.Entry<TopicPartition, OffsetAndTimestamp> entry :parMap.entrySet()) {
            TopicPartition key = entry.getKey();
            OffsetAndTimestamp value = entry.getValue();
            if (key == null || value == null) continue;
            Long offset = value.offset();
            System.out.println("partition-" + key.partition() +"|offset-" + offset);
            System.out.println();
            //根据消费里的timestamp确定offset
            if (value != null) {
                consumer.assign(Arrays.asList(key));
                consumer.seek(key, offset);
            }
        }

        while (true) {
            /*
             * poll() API 是拉取消息的⻓轮询
             */
            ConsumerRecords<String, String> records =consumer.poll(Duration.ofMillis( 1000 ));
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("收到消息：partition = %d,offset = %d, key =%s, value = %s%n", record.partition(),record.offset(), record.key(), record.value());
            }


            if (records.count() > 0 && Objects.equals("sync",flag) ) {
                // 手动同步提交offset，当前线程会阻塞直到offset提交成功
                // 一般使用同步提交，因为提交之后一般也没有什么逻辑代码了
                log.info("同步提交了offset");
                consumer.commitSync();
            }

            if (records.count() > 0  && Objects.equals("async",flag)) {
                // 手动异步提交offset，当前线程提交offset不会阻塞，可以继续处理后面的程序逻辑
                consumer.commitAsync(new OffsetCommitCallback() {
                    @Override
                    public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
                        if (exception != null) {
                            System.err.println("Commit failed for " + offsets);
                            System.err.println("Commit failed exception: " +exception.getStackTrace());
                        }
                    }
                });
                log.info("异步提交了offset");
            }

        }
    }
}
